package org.robby.hbase;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.NavigableMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.robby.hbase.entity.Post;

public class HBaseIF {
	Configuration conf = null;

	private static String tab_global = "tab_global";
	private static String tab_user2id = "tab_user2id";
	private static String tab_id2user = "tab_id2user";
	private static String tab_follow = "tab_follow";
	private static String tab_followed = "tab_followed";
	private static String tab_post = "tab_post";
	private static String tab_inbox = "tab_inbox";
	private static String tab_cdr = "tab_cdr";

	public HBaseIF() {
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "192.168.1.90");// 连接到远程服务上
	}

	public void createTable(String tbName, String colName, Integer ver)
			throws Exception {
		HBaseAdmin admin = new HBaseAdmin(conf);
		if (admin.tableExists(tbName)) {
			admin.disableTable(tbName);
			admin.deleteTable(tbName);
		}
		HTableDescriptor tableDes = new HTableDescriptor(tbName);
		HColumnDescriptor hd = new HColumnDescriptor(colName);
		hd.setMaxVersions(ver);
		tableDes.addFamily(hd);
		admin.createTable(tableDes);
	}

	// tab_global param:userid
	// tab_user2id info:id
	// tab_id2user info:username info:password
	// tab_follow name:userid
	// tab_followed userid

	public void createTables() throws Exception {
		createTable(tab_global, "param", 1);
		Put put = new Put(Bytes.toBytes("row_userid"));
		long id = 0;
		put.add(Bytes.toBytes("param"), Bytes.toBytes("userid"),
				Bytes.toBytes(id));
		HTable ht = new HTable(conf, tab_global);
		ht.put(put);

		createTable(tab_user2id, "info", 1);
		createTable(tab_id2user, "info", 1);
		createTable(tab_follow, "name", 1);
		createTable(tab_followed, "userid", 1);
		createTable(tab_cdr, "data", 1);
		// tab_post
		// rowkey:postid
		// CF:post:username,post:content,post:ts
		createTable(tab_post, "post", 1);
		put = new Put(Bytes.toBytes("row_postid"));
		id = 0;
		put.add(Bytes.toBytes("param"), Bytes.toBytes("postid"),
				Bytes.toBytes(id));
		ht.put(put);

		// tab_inbox
		// rowkey:userid_postid
		// CF:postid
		createTable(tab_inbox, "postid", 1);
	}

	public boolean post(String username, String content) throws Exception {
		HTable global = new HTable(conf, tab_global);
		HTable post = new HTable(conf, tab_post);
		long postid = global.incrementColumnValue(Bytes.toBytes("row_postid"),
				Bytes.toBytes("param"), Bytes.toBytes("postid"), 1l);
		DateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");
		String ts = df.format(new Date());
		Put put = new Put(Bytes.toBytes(postid));
		put.add(Bytes.toBytes("post"), Bytes.toBytes("username"),
				Bytes.toBytes(username));
		put.add(Bytes.toBytes("post"), Bytes.toBytes("content"),
				Bytes.toBytes(content));
		put.add(Bytes.toBytes("post"), Bytes.toBytes("ts"), Bytes.toBytes(ts));
		post.put(put);
		global.close();
		post.close();
		// 取得当前发布微博的用户ID
		long posterId = this.getIdByUsername(username);
		// 拼装rowid的范围，
		byte[] begin = Bytes.add(Bytes.toBytes(posterId), Bytes.toBytes("_"),
				Bytes.toBytes(Long.MAX_VALUE - Long.MAX_VALUE));
		byte[] end = Bytes.add(Bytes.toBytes(posterId), Bytes.toBytes("_"),
				Bytes.toBytes(Long.MAX_VALUE));

		// scan根据rowid的范围 进行扫描表
		Scan s = new Scan();
		s.setStartRow(begin);
		s.setStopRow(end);

		HTable followed = new HTable(conf, tab_followed);
		HTable inbox = new HTable(conf, tab_inbox);

		ResultScanner ss = followed.getScanner(s);

		List<Put> l = new ArrayList<Put>();
		// tab_inbox
		// 将发微博者的微博ID录到自己的收件箱中
		put = new Put(Bytes.add(Bytes.toBytes(posterId), Bytes.toBytes("_"),
				Bytes.toBytes(postid)));
		put.add(Bytes.toBytes("postid"), null, Bytes.toBytes(postid));
		// inbox.put(put);
		l.add(put);

		// 将发微笑博者的微博ID发送给关注自己的用户的收件箱中
		for (Result r : ss) {
			byte[] did = r.getValue("userid".getBytes(), null);
			put = new Put(Bytes.add(did, "_".getBytes(), Bytes.toBytes(postid)));
			put.add(Bytes.toBytes("postid"), null, Bytes.toBytes(postid));
			// inbox.put(put);
			l.add(put);
		}
		inbox.put(l);
		inbox.close();
		followed.close();

		return true;
	}

	public List<Post> getPostList(String username) throws Exception {
		List<Post> postList = new ArrayList<Post>();
		long id = this.getIdByUsername(username);
//		byte[] startRow = Bytes.add(Bytes.toBytes(id), "_".getBytes(),
//				Bytes.toBytes(Long.MAX_VALUE - Long.MAX_VALUE));
//		byte[] stopRow = Bytes.add(Bytes.toBytes(id), "_".getBytes(),
//				Bytes.toBytes(Long.MAX_VALUE));
		byte[] startRow = Bytes.toBytes(id);
		byte[] stopRow = Bytes.toBytes(id+1);
		Scan s = new Scan();
		s.setStartRow(startRow);
		s.setStopRow(stopRow);
		HTable inbox = new HTable(conf, tab_inbox);
		HTable post = new HTable(conf, tab_post);

		ResultScanner ss = inbox.getScanner(s);
		Get get = null;
		Result rs = null;
		Post p = null;
		String postUserName = null, content = null, ts = null;
		for (Result r : ss) {
			byte[] postid = r.getValue("postid".getBytes(), null);
			get = new Get(postid);
			rs = post.get(get);
			postUserName = Bytes.toString(rs.getValue("post".getBytes(),
					"username".getBytes()));
			content = Bytes.toString(rs.getValue("post".getBytes(),
					"content".getBytes()));
			ts = Bytes
					.toString(rs.getValue("post".getBytes(), "ts".getBytes()));
			postList.add(new Post(postUserName, content, ts));
		}
		return postList;
	}

	// 1.check user exist
	// 2.get new ID
	// 3. insert tab_user2id
	// 4.insert tab_id2user
	public boolean createNewUser(String username, String password)
			throws Exception {
		HTable htab_user2id = new HTable(conf, tab_user2id);
		HTable htab_id2user = new HTable(conf, tab_id2user);
		HTable htab_global = new HTable(conf, tab_global);
		if (htab_user2id.exists(new Get(username.getBytes()))) {
			return false;
		}
		long id = htab_global.incrementColumnValue(Bytes.toBytes("row_userid"),
				Bytes.toBytes("param"), Bytes.toBytes("userid"), 1);
		Put put = new Put(username.getBytes());
		put.add("info".getBytes(), "id".getBytes(), Bytes.toBytes(id));
		htab_user2id.put(put);

		put = new Put(Bytes.toBytes(id));
		put.add("info".getBytes(), "username".getBytes(),
				Bytes.toBytes(username));
		put.add("info".getBytes(), "password".getBytes(),
				Bytes.toBytes(password));
		htab_id2user.put(put);

		return true;
	}
	
	
	public void getFamilyMap(String rowKey) throws Exception {
		HTable htab_id2user = new HTable(conf, tab_id2user);
		Get get = new Get();
		get.setId(rowKey);
		Result rs = htab_id2user.get(get);
		NavigableMap<byte[], byte[]> map = rs.getFamilyMap("info".getBytes());
		Iterator<byte[]> itKey = map.keySet().iterator();
		while(itKey.hasNext()){
			byte[] key = itKey.next();
			byte[] value = map.get(key);
			System.out.println("key:"+Bytes.toString(key)+",value:"+Bytes.toString(value));
		}
		
	}

	/**
	 * 通过用户名取得ID
	 * 
	 * @param username
	 * @return
	 * @throws Exception
	 */
	public long getIdByUsername(String username) throws Exception {
		HTable ht = null;
		try {
			ht = new HTable(conf, tab_user2id);
			Result rs = ht.get(new Get(username.getBytes()));
			KeyValue kv = rs
					.getColumnLatest("info".getBytes(), "id".getBytes());
			if (kv != null) {
				return Bytes.toLong(kv.getValue());
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			ht.close();
		}
		return 0;
	}

	/**
	 * 通过ID取得ID
	 * 
	 * @param id
	 * @return
	 * @throws Exception
	 */
	public String getUsernameById(long id) throws Exception {
		HTable ht = null;
		try {
			ht = new HTable(conf, tab_id2user);
			Result rs = ht.get(new Get(Bytes.toBytes(id)));
			KeyValue kv = rs.getColumnLatest("info".getBytes(),
					"username".getBytes());
			return Bytes.toString(kv.getValue());
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			ht.close();
		}
		return null;
	}

	/**
	 * 检查密码
	 * 
	 * @param username
	 * @param password
	 * @return
	 * @throws Exception
	 */
	public long checkPassword(String username, String password)
			throws Exception {
		try {
			long id = getIdByUsername(username);
			if (id > 0) {
				HTable ht = new HTable(conf, tab_id2user);
				Result rs = ht.get(new Get(Bytes.toBytes(id)));
				KeyValue kv = rs.getColumnLatest("info".getBytes(),
						"password".getBytes());
				return Bytes.toLong(kv.getKey());
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return 0;
	}

	/**
	 * 
	 * @param ownerName
	 *            关注者
	 * @param distName
	 *            被关注者
	 * @return
	 * @throws Exception
	 */
	public boolean follow(String ownerName, String distName) throws Exception {
		long ownerId = this.getIdByUsername(ownerName);
		long distId = this.getIdByUsername(distName);
		if (ownerId == 0 || distId == 0 || ownerId == distId) {
			return false;
		}
		// tab_follow
		HTable follow = new HTable(conf, tab_follow);
		Put put = new Put(Bytes.toBytes(ownerId));
		put.add(Bytes.toBytes("name"), Bytes.toBytes(distId),
				distName.getBytes());
		follow.put(put);
		follow.close();
		// tab_followed
		HTable followed = new HTable(conf, tab_followed);
		put = new Put(Bytes.add(Bytes.toBytes(distId), "_".getBytes(),
				Bytes.toBytes(ownerId)));// row key
		put.add("userid".getBytes(), null, Bytes.toBytes(ownerId));
		followed.put(put);
		followed.close();
		return true;
	}

	/**
	 * 
	 * @param ownerName
	 *            发起关注者
	 * @param distName
	 *            被关注者
	 * @return
	 * @throws Excpetion
	 */
	public boolean unFollow(String ownerName, String distName) throws Exception {
		long ownerId = this.getIdByUsername(ownerName);
		long distId = this.getIdByUsername(distName);
		if (ownerId == 0 || distId == 0 || ownerId == distId) {
			return false;
		}
		HTable follow = new HTable(conf, tab_follow);
		Delete del = new Delete(Bytes.toBytes(ownerId));
		del.deleteColumns(Bytes.toBytes("name"), Bytes.toBytes(distId)); // 删除某个qulifier的所有版本数据
		follow.delete(del);
		follow.close();
		HTable followed = new HTable(conf, tab_followed);
		del = new Delete(Bytes.add(Bytes.toBytes(distId), "_".getBytes(),
				Bytes.toBytes(ownerId)));
		followed.delete(del);
		followed.close();
		return true;
	}

	public List<String> getAllUsers() throws Exception {
		List<String> users = new ArrayList<String>();
		HTable user2id = new HTable(conf, tab_user2id);
		Scan s = new Scan();
		ResultScanner ss = user2id.getScanner(s);
		for (Result r : ss) {
			String name = new String(r.getRow());
			users.add(name);
		}
		user2id.close();
		return users;
	}

	/**
	 * 取得关注的用户
	 * 
	 * @param ownerName
	 * @return
	 * @throws Exception
	 */
	public List<String> getFollow(String ownerName) throws Exception {
		List<String> followUsers = new ArrayList<String>();
		long ownerId = this.getIdByUsername(ownerName);
		HTable follow = new HTable(conf, tab_follow);
		Get get = new Get(Bytes.toBytes(ownerId));
		Result rs = follow.get(get);
		for (KeyValue kv : rs.raw()) {
			followUsers.add(new String(kv.getValue()));
		}
		return followUsers;
	}

	/**
	 * 取得我被关注的列表
	 * 
	 * @param distName
	 * @return
	 * @throws Exception
	 */
	public List<String> getFollowed(String distName) throws Exception {
		List<String> users = new ArrayList<String>();
		long distId = this.getIdByUsername(distName);
		HTable followed = new HTable(conf, tab_followed);
		Scan s = new Scan(new Get(Bytes.toBytes(distId)));
		ResultScanner rss = followed.getScanner(s);
		Iterator<Result> itRss = rss.iterator();
		while (itRss.hasNext()) {
			Result rs = itRss.next();
			rs.getRow();
		}
		return users;
	}

	public static void main(String[] args) {
		HBaseIF hif = new HBaseIF();
		try {
			hif.createTable(tab_cdr, "data", 1);
//			hif.createTables();
			// hif.createNewUser("user1", "pwd1");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
